package ex.datastream.functions.richFunction;

import ex.vo.KeyedState;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
import org.apache.flink.util.Collector;

import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

public class KeyedStateReaderFunc extends KeyedStateReaderFunction<String, KeyedState> {
	  ValueState<Tuple2<String, Integer>>  state;
	 
	  ListState<Long> updateTimes;

	  @Override
	  public void open(Configuration parameters) {

		ValueStateDescriptor descriptor = new ValueStateDescriptor("state",
				  TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {}));
	    state = getRuntimeContext().getState(descriptor);

	    ListStateDescriptor<Long> updateDescriptor = new ListStateDescriptor<>("times", Types.LONG);
	    updateTimes = getRuntimeContext().getListState(updateDescriptor);
	  }
	 
	  @Override
	  public void readKey(
	    String key,
	    Context ctx,
	    Collector<KeyedState> out) throws Exception {
	        
	    KeyedState data = new KeyedState();
	    data.key    = key;
	    data.value  = state.value();
	    data.times  = StreamSupport
	      .stream(updateTimes.get().spliterator(), false)
	      .collect(Collectors.toList());

	    out.collect(data);
	  }
}
